package com.lagou.client;

import com.lagou.encoder.JSONSerializer;
import com.lagou.encoder.RpcEncoder;
import com.lagou.request.RpcRequest;
import com.lagou.service.UserService;
import com.lagou.zk.ServerInfoVO;
import io.netty.bootstrap.Bootstrap;
import io.netty.channel.ChannelInitializer;
import io.netty.channel.ChannelOption;
import io.netty.channel.ChannelPipeline;
import io.netty.channel.EventLoopGroup;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.SocketChannel;
import io.netty.channel.socket.nio.NioSocketChannel;
import io.netty.handler.codec.string.StringDecoder;
import lombok.extern.slf4j.Slf4j;

import java.lang.reflect.InvocationHandler;
import java.lang.reflect.Method;
import java.lang.reflect.Proxy;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.atomic.AtomicInteger;

/**
 * 客户端线程
 */
@Slf4j
public class RpcConsumer {
    public static final AtomicInteger REQUEST_ID_SEQ = new AtomicInteger();
    //创建线程池对象
    private static ExecutorService executor = Executors.newFixedThreadPool(Runtime.getRuntime().availableProcessors());
    private ServerInfoVO serverInfo;
    private EventLoopGroup group;
    private UserService service;

    //2.初始化netty客户端
    public static RpcConsumer create(ServerInfoVO serverInfo) throws InterruptedException {
        RpcConsumer consumer = new RpcConsumer();
        consumer.init(serverInfo);
        return consumer;
    }

    public ServerInfoVO getServerInfo() {
        return serverInfo;
    }

    public UserService getService() {
        return service;
    }

    /**
     * 关闭连接
     */
    public void shutdown() {
        try {
            // 当线程状态为DOWN时判断连接
            this.group.shutdownGracefully();
        } catch (Exception e) {
            log.error(e.getMessage());
        }
        log.info("关闭与服务[{}@{}:{}]的连接", serverInfo.getServerName(), serverInfo.getIp(), serverInfo.getPort());
    }

    /**
     * 初始化客户端并建立连接
     *
     * @param serverInfo
     * @throws InterruptedException
     */
    private void init(ServerInfoVO serverInfo) throws InterruptedException {
        UserClientHandler userClientHandler = new UserClientHandler();
        this.group = new NioEventLoopGroup();
        Bootstrap connection = new Bootstrap();
        this.serverInfo = serverInfo;
        connection.group(group)
                .channel(NioSocketChannel.class)
                .option(ChannelOption.TCP_NODELAY, true)
                .option(ChannelOption.CONNECT_TIMEOUT_MILLIS, 1000)
                .handler(new ChannelInitializer<SocketChannel>() {
                    protected void initChannel(SocketChannel ch) throws Exception {
                        ChannelPipeline pipeline = ch.pipeline();
                        // 绑定出参格式为RpcRequest
                        pipeline.addLast(new RpcEncoder<>(RpcRequest.class, new JSONSerializer()));
                        // 绑定入参为String
                        pipeline.addLast(new StringDecoder());
                        pipeline.addLast(userClientHandler);
                    }
                });
        connection.connect(serverInfo.getIp(), serverInfo.getPort()).sync();
        this.service = createProxy(UserService.class, userClientHandler);
    }

    //1.创建一个代理对象 providerName：UserService#sayHello are you ok?
    @SuppressWarnings("unchecked")
    private <T> T createProxy(final Class<T> serviceClass, UserClientHandler userClientHandler) {
        //借助JDK动态代理生成代理对象
        Object proxyObject = Proxy.newProxyInstance(Thread.currentThread().getContextClassLoader(), new Class<?>[]{serviceClass}, new InvocationHandler() {
            public Object invoke(Object proxy, Method method, Object[] args) throws Throwable {
                RpcRequest request = new RpcRequest();
                String requestId = String.format("%010d", REQUEST_ID_SEQ.addAndGet(1));
                request.setRequestId(requestId);
                request.setClassName(serviceClass.getSimpleName());
                request.setMethodName(method.getName());
                request.setParameters(args);
                request.setParameterTypes(new Class[]{String.class});
                // 设置参数
                userClientHandler.setPara(request);
                // 去服务端请求数据
                return executor.submit(userClientHandler).get();
            }
        });
        return (T) proxyObject;
    }

    public String send(String message) {
        String sendResult = getService().sayHello(message);
        log.info("Send to [{}@{}:{}] = {}", getServerInfo().getServerName(), getServerInfo().getIp(), getServerInfo().getPort(), sendResult);
        return sendResult;
    }
}
